#include "NetMgr.h"
#include <forward_list>
#include <sstream>
#include <random>
#include <string>
#include <array>
#include <stdint.h>

NetMgr *NetMgr::m_oInstance = NULL;
std::once_flag NetMgr::m_flag;

class EventN
{
public:
	int32_t Wait()
	{
		std::unique_lock<decltype(m_MutexCondition)> lock{ m_MutexCondition };
		m_Condition.wait(lock, [&]()->bool { return ret != -1; }); // suspend and wait ...	
		return ret;
	}
	void Set()
	{
		std::unique_lock<decltype(m_MutexCondition)> lock(m_MutexCondition);
		ret = 0;
		m_Condition.notify_one();
	}
private:
	std::mutex m_MutexCondition;
	std::condition_variable m_Condition;
	int32_t ret = -1;
};

WorkerThreadN::WorkerThreadN()
{
}
WorkerThreadN::~WorkerThreadN()
{
	StopRun();
}

int32_t WorkerThreadN::UpdateChannel(ChannelPtr channel)
{

	//printf("### EventLoopN WorkerThreadN::UpdateChannel Start channel:%d ThreadID:%s this:%p \n", channel->GetSocket(), GetThreadId().c_str(), this);
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### WorkerThreadN::UpdateChannel Start channel:%d ThreadID:%s this:%p ", channel->GetSocket(), GetThreadId().c_str(), this);
	int32_t ret = -1;
	InvokTask([this, &channel, &ret]()
		{
			//printf("*** EventLoopN WorkerThreadN::UpdateChannel InvokTask Start  channel:%d ThreadID:%s this:%p \n", channel->GetSocket(), GetThreadId().c_str(), this);
			//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** WorkerThreadN::UpdateChannel InvokTask Start  channel:%d ThreadID:%s this:%p ", channel->GetSocket(), GetThreadId().c_str(), this);
			socket_t socket = channel->GetSocket();

			std::unordered_map<socket_t, ChannelPtr>::iterator iter = channels_.find(socket);
			if (iter != channels_.end()) {
				if (channel->IsNoneEvent()) {

					channels_.erase(socket);
					ret = 1;
				}
				else {
					// 更新读写事件 暂时不考虑
					//iter->second->
					ret = 2;
				}
			}
			else {
				if (!channel->IsNoneEvent()) {
					channels_.emplace(socket, channel);
					ret = 0;
				}
				else
				{
					ret = 3;
				}
				//printf("channel :%d \n", channel->GetSocket());
			}

			//printf("*** EventLoopN WorkerThreadN::UpdateChannel InvokTask End  channel:%d ThreadID:%s ret:%d this:%p \n", channel->GetSocket(), GetThreadId().c_str(), ret, this);
			//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** WorkerThreadN::UpdateChannel InvokTask End  channel:%d ThreadID:%s ret:%d this:%p ", channel->GetSocket(), GetThreadId().c_str(), ret, this);
		}
	);
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### WorkerThreadN::UpdateChannel End channel:%d ThreadID:%s ret:%d this:%p ", channel->GetSocket(), GetThreadId().c_str(), ret, this);
	//printf("### EventLoopN WorkerThreadN::UpdateChannel End channel:%d ThreadID:%s ret:%d this:%p \n", channel->GetSocket(), GetThreadId().c_str(), ret, this);
	return ret;
}

int32_t WorkerThreadN::RemoveChannel(ChannelPtr& channel)
{
	//printf("### EventLoopN WorkerThreadN::RemoveChannel Start channel:%d ThreadID:%s this:%p \n", channel->GetSocket(), GetThreadId().c_str(), this);
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### WorkerThreadN::RemoveChannel Start channel:%d ThreadID:%s this:%p ", channel->GetSocket(), GetThreadId().c_str(), this);
	int ret = -1;
	InvokTask([&channel, &ret, this]()
		{
			//printf("*** EventLoopN WorkerThreadN::RemoveChannel InvokTask Start  channel:%d ThreadID:%s this:%p \n", channel->GetSocket(), GetThreadId().c_str(), this);
			//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** WorkerThreadN::RemoveChannel InvokTask Start  channel:%d ThreadID:%s this:%p ", channel->GetSocket(), GetThreadId().c_str(), this);

			socket_t fd = channel->GetSocket();

			if (channels_.find(fd) != channels_.end()) {
				channels_.erase(fd);
				ret = 0;
			}
			else
			{
				ret = 1;
			}
			//printf("*** EventLoopN WorkerThreadN::RemoveChannel InvokTask End channel:%d ThreadID:%s ret:%d this:%p \n", channel->GetSocket(), GetThreadId().c_str(), ret, this);
			//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** WorkerThreadN::RemoveChannel InvokTask End channel:%d ThreadID:%s ret:%d this:%p ", channel->GetSocket(), GetThreadId().c_str(), ret, this);
		}
	);
	//printf("### EventLoopN WorkerThreadN::RemoveChannel End channel:%d ThreadID:%s ret:%d this:%p \n", channel->GetSocket(), GetThreadId().c_str(), ret, this);
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### WorkerThreadN::RemoveChannel End channel:%d ThreadID:%s ret:%d this:%p ", channel->GetSocket(), GetThreadId().c_str(), ret, this);
	return ret;
}

void WorkerThreadN::PostTask(Task task)
{
	// 调用线程和将要打入的线程是同一个
	if (_loopThreadId == std::this_thread::get_id())
	{
		// 原地执行就可以
		task();
		return;
	}

	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([this, &task]() {

			task();

			});
	}
	char event = 1;
	m_oPipe.Write(&event, 1);
}


void WorkerThreadN::InvokTask(Task task)
{
	// 调用线程和将要打入的线程是同一个
	if (_loopThreadId == std::this_thread::get_id())
	{
		// 原地执行就可以
		printf("##### yuan di zhi xing ok \n");
		task();
		return;
	}

	EventN tEvent;
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([task, &tEvent]() {
			task();
			tEvent.Set();
			});
	}
	char event = 1;
	m_oPipe.Write(&event, 1);
	//printf("Wait before \n");
	tEvent.Wait();
}

int64_t WorkerThreadN::AddTimer(int64_t iDelayMs, std::function<bool()> fcb)
{

	auto t_timer = std::make_shared<TimerNet>(iDelayMs, fcb);
	int64_t t_timer_id = 0;
	InvokTask([this, &t_timer, &t_timer_id]()
		{
			printf("***  WorkerThreadN::AddTimer InvokTask Start   ThreadID:%s this:%p \n",  GetThreadId().c_str(), this);
			

			m_iTimerId++;
			t_timer->m_iTimerId = m_iTimerId;
			t_timer_id = m_iTimerId;
			uint64_t time_line = getCurrentTicket() + t_timer->m_iDelay;
			m_delayTask.emplace(time_line, std::move(t_timer));

			printf("*** WorkerThreadN::AddTimer InvokTask End ThreadID:%s this:%p \n",  GetThreadId().c_str(), this);
			
		}
	);

	printf("###  WorkerThreadN::AddTimer End iDelay:%lld ThreadID:%s t_timer_id:%lld this:%p \n", iDelayMs, GetThreadId().c_str(), t_timer_id, this);
	
	return t_timer_id;
}

int32_t WorkerThreadN::DelTimer(int64_t timerId)
{
	int ret = -1;
	InvokTask([this, timerId, &ret]()
		{
			printf("*** WorkerThreadN WorkerThreadN::DelTimer Start InvokTask timerId:%lld ThreadID:%s this:%p \n", timerId, GetThreadId().c_str(), this);
			
			std::multimap<uint64_t, std::shared_ptr<TimerNet> >::iterator it = m_delayTask.begin();
			for (; it != m_delayTask.end();)
			{
				std::shared_ptr<TimerNet> pTimer = it->second;
				if (pTimer->m_iTimerId == timerId)
				{
					it = m_delayTask.erase(it);
					ret = 0;
					break;
				}
				else
				{
					it++;
				}
			}

			printf("*** WorkerThreadN WorkerThreadN::DelTimer End InvokTask timerId:%lld ThreadID:%s ret:%d this:%p \n", timerId, GetThreadId().c_str(), ret, this);
			

		});

	printf("### WorkerThreadN WorkerThreadN::DelTimer End timerId:%lld ThreadID:%s ret:%d this:%p \n", timerId, GetThreadId().c_str(), ret, this);
	
	return ret;
}


int64_t WorkerThreadN::getMinDelay() {
	auto it = m_delayTask.begin();
	if (it == m_delayTask.end()) {
		return 100;
	}
	auto now = getCurrentTicket();
	if (it->first > now) {

		////printf("it->first:%llu now:%lld dif:%lld \n", it->first, now, it->first - now);
		return it->first - now;
	}

	return flushDelayTask(now);
}

int64_t WorkerThreadN::flushDelayTask(uint64_t now_time) {

	////printf("############# sleep :%llu \n", now_time);
	decltype(m_delayTask) taskCopy;
	taskCopy.swap(m_delayTask);
	uint64_t start = getCurrentTicket();
	for (auto it = taskCopy.begin(); it != taskCopy.end() && it->first <= now_time; it = taskCopy.erase(it)) {

		//auto next_delay = (*(it->second))();
		std::shared_ptr<TimerNet>  pTimer = it->second;
		if (pTimer->m_cb)
		{
			bool ret = pTimer->m_cb();
			if (ret)
			{
				m_delayTask.emplace(pTimer->m_iDelay + now_time, std::move(it->second));
			}
			else
			{
				//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "TaskScheduler::flushDelayTask this:%p m_iTimerId:%lld ret:%d", this, pTimer->m_iTimerId, ret);
			}
		}
	}
	uint64_t end = getCurrentTicket();
	int64_t useTime = end - start;
	if (useTime > 11)
	{
		////printf("timers_poll::flushDelayTask useTimer = %d \n", useTime);
	}

	taskCopy.insert(m_delayTask.begin(), m_delayTask.end());
	taskCopy.swap(m_delayTask);

	auto it = m_delayTask.begin();
	if (it == m_delayTask.end()) {

		return 0;
	}


	int64_t t_sleep = it->first - getCurrentTicket();
	if (t_sleep < -1)
	{
		//printf("timers_poll::flushDelayTask  t_sleep = %lld  \n", t_sleep);
	}
	return t_sleep > 0 ? t_sleep : 0;

}
std::string WorkerThreadN::GetThreadId()
{
	std::ostringstream oss;
	oss << _loopThreadId;
	return oss.str();
}
int64_t WorkerThreadN::getCurrentTicket()
{
	//std::chrono::steady_clock::time_point t1 = std::chrono::steady_clock::now();
	//std::chrono::steady_clock::time_point t11;
	//int64_t duration = std::chrono::duration_cast<std::chrono::microseconds>(t1 - t11).count();
	//int64_t duration = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t11).count();
	return  std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
}

void WorkerThreadN::SetThreadName(std::string threadName)
{
#ifdef WIN32

	int wcharCount = MultiByteToWideChar(CP_UTF8, 0, threadName.c_str(), -1, NULL, 0);
	std::wstring wideString(wcharCount, L'\0');
	MultiByteToWideChar(CP_UTF8, 0, threadName.c_str(), -1, &wideString[0], wcharCount);

	PCWSTR pwstr = wideString.c_str();
	HRESULT hr = SetThreadDescription(m_loopThread->native_handle(), pwstr);
	if (FAILED(hr))
	{
		//MEDIA_TRACE_LOG("NetMgr", LL_ERROR, "*** WorkerThreadN::SetThreadName error this:%p ", this);
	}
#else
	pthread_setname_np(m_loopThread->native_handle(), threadName.c_str());
#endif
}
int32_t WorkerThreadN::StartRun(std::string threadName)
{

	int ret = m_oPipe.Create();
	if (0 != ret)
	{
		return -2;
	}

	m_loopThread = new std::thread(&WorkerThreadN::RunLoop, this);
	while (!m_bStart)
	{
		std::this_thread::sleep_for(std::chrono::milliseconds(1));
	}
	//std::unique_lock<std::mutex> lock(_mutex);
	//_condition.wait(lock, [this]() { return m_bStart == true; });
	SetThreadName(threadName);
	if (NULL == m_loopThread)
	{
		return -1;
	}
	
	return 0;

}
void WorkerThreadN::StopRun()
{
	m_bExitThread = true;

	// 驱动线程快速退出
	char event = 1;
	m_oPipe.Write(&event, 1);

	if (m_loopThread)
	{
		m_loopThread->join();
		delete m_loopThread;
		m_loopThread = 0;
	}

	m_oPipe.Close();
}



void WorkerThreadN::RunLoop()
{
	_loopThreadId = std::this_thread::get_id();
	m_bStart = true;
	//_condition.notify_one();

	fd_set fd_read;
	fd_set fd_write;
	fd_set fd_exp;


	while (!m_bExitThread)
	{
		int max_fd = 0;

		FD_ZERO(&fd_read);
		FD_ZERO(&fd_write);
		FD_ZERO(&fd_exp);


		FD_SET(m_oPipe.Read(), &fd_read);
		max_fd = m_oPipe.Read();

		for (auto iter : channels_) {
			int events = iter.second->GetEvents();
			socket_t fd = iter.second->GetSocket();

			if ((events & EVENT_IN)) {
				FD_SET(fd, &fd_read);
			}

			if ((events & EVENT_OUT)) {
				FD_SET(fd, &fd_write);
			}
			if (fd > max_fd)
			{
				max_fd = fd;
			}

		}
		struct timeval tv;
		int32_t minDelay = getMinDelay();
		tv.tv_sec = minDelay / 1000;
		tv.tv_usec = 1000 * (minDelay % 1000);
		int ret = select(max_fd + 1, &fd_read, &fd_write, NULL, &tv);

		if (ret > 0)
		{

			std::forward_list<std::pair<ChannelPtr, int>> event_list;
			 {

				for (auto iter : channels_) {
					int events = 0;
					socket_t socket = iter.second->GetSocket();

					if (FD_ISSET(socket, &fd_read)) {
						events |= EVENT_IN;
					}

					if (FD_ISSET(socket, &fd_write)) {
						events |= EVENT_OUT;
					}

					if (FD_ISSET(socket, &fd_exp)) {
						events |= (EVENT_HUP); // close
					}

					if (events != 0) {
						event_list.emplace_front(iter.second, events);
					}
				}
			}

			for (auto& iter : event_list) {
				iter.first->HandleEvent(iter.second);
			}
			
			if (FD_ISSET(m_oPipe.Read(), &fd_read))
			{
				char buf[1];
				m_oPipe.Read(buf, sizeof(buf));		

				// 处理增删任务
				std::list<Task> t_queue;
				{
					std::unique_lock<std::mutex> lock(_mutex);
					if (!_queue.empty())
					{
						t_queue.swap(_queue);

					}
				}

				// 处理任务
				if (!t_queue.empty())
				{
					for (std::list<Task>::iterator it = t_queue.begin(); it != t_queue.end(); it++)
					{
						Task task = *it;
						task();
					}
					t_queue.clear();
					//printf("处理管道任务\n");
				}
			}
		}

	}

}

EventLoopN::EventLoopN()
{

}

EventLoopN::~EventLoopN()
{
	Quit();
}

int32_t EventLoopN::Loop(std::string name_thread, uint32_t num_threads)
{
	std::lock_guard<std::mutex> locker(mutex_);
	num_threads_ = num_threads;
	m_name = name_thread;
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### EventLoopN::Loop Start num_threads:%d name:%s this:%p ", num_threads, m_name.c_str(), this);
	if (!task_schedulers_.empty()) {
		//printf("### EventLoopN Have Looped !task_schedulers_.empty() this:%p task_schedulers_.size():%d \n", this, task_schedulers_.size());
		//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### EventLoopN End Have Looped !task_schedulers_.empty() this:%p task_schedulers_.size():%d ", this, task_schedulers_.size());
		return task_schedulers_.size();
	}
	for (int32_t n = 0; n < num_threads_; n++)
	{
		int32_t ret = 0;
		//printf("*** EventLoopN::Loop StartRun before n:%d this:%p num_threads_:%d \n", n, this, num_threads_);
		//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** EventLoopN::Loop StartRun before n:%d this:%p ", n, this);
		std::string t_thread_name = m_name + "_" + std::to_string(n);
		std::shared_ptr<WorkerThreadN> task_scheduler_ptr(new WorkerThreadN());
		ret = task_scheduler_ptr->StartRun(t_thread_name);

		//printf("*** EventLoopN::Loop StartRun end n:%d ret:%d task_scheduler_ptr:%p ThreadID:%s this:%p \n", n, ret, task_scheduler_ptr.get(), task_scheduler_ptr->GetThreadId().c_str(), this);
		//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** EventLoopN::Loop StartRun end n:%d ret:%d task_scheduler_ptr:%p ThreadID:%s this:%p ", n, ret, task_scheduler_ptr.get(), task_scheduler_ptr->GetThreadId().c_str(), this);
		if (0 == ret)
		{
			task_schedulers_.push_back(task_scheduler_ptr);
		}
	
	}
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### EventLoopN::Loop End task_schedulers_.size():%d this:%p ", task_schedulers_.size(), this);
	return task_schedulers_.size();
}

void EventLoopN::Quit()
{
	//printf("EventLoopN::Quit Start this:%p task_schedulers_.size():%d \n", this, task_schedulers_.size());
	////MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### EventLoopN::Quit Start this:%p task_schedulers_.size():%d ", this, task_schedulers_.size());
	int i = 0;
	for (auto iter : task_schedulers_)
	{
		std::shared_ptr<WorkerThreadN> t_scheduler = iter;
		//printf("EventLoopN::StopRun before i:%d iter:%p ThreadID:%s this:%p \n", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		////MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** EventLoopN::StopRun before i:%d iter:%p ThreadID:%s this:%p ", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		iter->StopRun();
		//printf("EventLoopN::StopRun after i:%d iter:%p ThreadID:%s this:%p \n", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		////MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** EventLoopN::StopRun after i:%d iter:%p ThreadID:%s this:%p ", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		i++;
	}
	task_schedulers_.clear();
	//printf("EventLoopN::Quit End this:%p \n", this);
	////MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### EventLoopN::Quit End this:%p ", this);
}

std::shared_ptr<WorkerThreadN> EventLoopN::GetTaskScheduler()
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() == 1) {
		return task_schedulers_.at(0);
	}
	else {
		auto task_scheduler = task_schedulers_.at(index_);
		index_++;
		if (index_ >= task_schedulers_.size()) {
			index_ = 0;
		}
		return task_scheduler;
	}

	return nullptr;
}

void EventLoopN::UpdateChannel(ChannelPtr channel)
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() > 0) {
		task_schedulers_[0]->UpdateChannel(channel);
	}
}

void EventLoopN::RemoveChannel(ChannelPtr& channel)
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() > 0) {
		task_schedulers_[0]->RemoveChannel(channel);
	}
}

NetMgr::NetMgr()
{
}

NetMgr* NetMgr::GetInstance()
{
	//return &(NetMgr::m_oInstance);
	std::call_once(m_flag, [&](){
		NetMgr::m_oInstance = new NetMgr();
		printf("\n *********** ########## NetMgr::GetInstance call_once \n");
	});
	if (NetMgr::m_oInstance)
	{
		return NetMgr::m_oInstance;
	}
	return NULL;
}

int32_t NetMgr::Init(uint32_t num_threads, std::string name_thread)
{
	int32_t ret = m_oEventLoop.Loop(name_thread, num_threads);
	return ret;
}

NetMgr::~NetMgr()
{
	m_oEventLoop.Quit();
}



Pipe::Pipe()
{

}

Pipe::~Pipe()
{
	Close();
}

void SetNonBlock(socket_t fd)
{
#if defined(__linux) || defined(__linux__) 
	int flags = fcntl(fd, F_GETFL, 0);
	fcntl(fd, F_SETFL, flags | O_NONBLOCK);
#else
	unsigned long on = 1;
	ioctlsocket(fd, FIONBIO, &on);
#endif
}

void SetBlock(socket_t fd, int write_timeout)
{
#if defined(__linux) || defined(__linux__) 
	int flags = fcntl(fd, F_GETFL, 0);
	fcntl(fd, F_SETFL, flags & (~O_NONBLOCK));
#elif defined(WIN32) || defined(_WIN32)
	unsigned long on = 0;
	ioctlsocket(fd, FIONBIO, &on);
#else
#endif
	if (write_timeout > 0)
	{
#ifdef SO_SNDTIMEO
#if defined(__linux) || defined(__linux__) 
		struct timeval tv = { write_timeout / 1000, (write_timeout % 1000) * 1000 };
		setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, (char*)&tv, sizeof tv);
#elif defined(WIN32) || defined(_WIN32)
		unsigned long ms = (unsigned long)write_timeout;
		setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, (char*)&ms, sizeof(unsigned long));
#else
#endif		
#endif
	}
}


static uint16_t get_local_port(int fd) {
	struct sockaddr addr;
	struct sockaddr_in* addr_v4;
	socklen_t addr_len = sizeof(addr);
	//获取remote ip and port
	if (0 == getsockname(fd, &addr, &addr_len)) {
		if (addr.sa_family == AF_INET) {
			addr_v4 = (sockaddr_in*)&addr;
			return ntohs(addr_v4->sin_port);
		}
	}
	return 0;
}
int Pipe::Create()
{
#if defined(WIN32) || defined(_WIN32) 
	//AfTcpSocket rp(socket(AF_INET, SOCK_STREAM, 0)), wp(socket(AF_INET, SOCK_STREAM, 0));
	AfTcpSocket rp;
	AfTcpSocket wp;
	int ret = 0;

	std::random_device rd;
	uint16_t port = 0;
	int again = 5;

	while (again--) {
		//port = rd(); 
		if (rp.Open(AfSockAddr("127.0.0.1", port), true) == 0) {
			port = get_local_port(rp.hSock);
			break;
		}
	}

	if (again == 0) {
		return -1;
	}

	if (rp.Listen(1) < 0) {
		return -2;
	}
	if (0 != wp.Open())
	{
		return -3;
	}
	if (0 != wp.Connect(AfSockAddr("127.0.0.1", port))) {
		return -4;
	}
	pipe_fd_[1] = wp.hSock;
	AfTcpSocket p;
	ret = rp.Accept(&p);
	if (ret < 0) {
		return -5;
	}
	pipe_fd_[0] = p.hSock;

	SetNonBlock(pipe_fd_[0]);
	SetNonBlock(pipe_fd_[1]);
#elif defined(__linux) || defined(__linux__) 
	if (pipe2(pipe_fd_, O_NONBLOCK | O_CLOEXEC) < 0) {
		return -1;
	}
#endif
	return 0;
}

int Pipe::Write(void* buf, int len)
{
#if defined(WIN32) || defined(_WIN32) 
	return ::send(pipe_fd_[1], (char*)buf, len, 0);
#elif defined(__linux) || defined(__linux__) 
	return ::write(pipe_fd_[1], buf, len);
#endif 
}

int Pipe::Read(void* buf, int len)
{
#if defined(WIN32) || defined(_WIN32) 
	return recv(pipe_fd_[0], (char*)buf, len, 0);
#elif defined(__linux) || defined(__linux__) 
	return ::read(pipe_fd_[0], buf, len);
#endif 
}

void Pipe::Close()
{
#if defined(WIN32) || defined(_WIN32) 
	closesocket(pipe_fd_[0]);
	closesocket(pipe_fd_[1]);
#elif defined(__linux) || defined(__linux__) 
	::close(pipe_fd_[0]);
	::close(pipe_fd_[1]);
#endif

}